package net.bw.realtime.tmall.ods.job;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.javafaker.Faker;
import net.bw.realtime.tmall.common.utils.KafkaUtil;
import net.bw.realtime.tmall.ods.bean.TmallLogBean;
import net.bw.realtime.tmall.ods.utils.RandomTimestampToday;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.HashMap;
import java.util.Locale;
import java.util.Map;
import java.util.Random;

import static net.bw.realtime.tmall.ods.utils.ChineseLocationUtils.getRandomCity;
import static net.bw.realtime.tmall.ods.utils.ChineseLocationUtils.getRandomProvince;

/*
 * @ Author：liuyawei
 * @ Date：2025-05-26
 */
public class TmallLogMockJobtest {

    public static void main(String[] args) throws Exception {

        // 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<TmallLogBean> tmallLogBeanDataStreamSource = env.addSource(
                new RichParallelSourceFunction<TmallLogBean>() {

                    private volatile boolean isRunning = true;
                    private final Random random = new Random();
                    private final String[] actionTypes = {"view_product", "add_to_cart", "place_order", "pay_success"};
                    private final String[] categories = {"手机数码", "家用电器", "服饰箱包", "美妆护肤", "食品饮料"};
                    private final String[] deviceTypes = {"PC", "无线"};

                    @Override
                    public void run(SourceContext<TmallLogBean> ctx) throws Exception {

                        int count = 0;

                        while (isRunning && count < 10000) {
                            TmallLogBean tmallLogBean = new TmallLogBean();

                            // 时间改为 Long 类型
                            RandomTimestampToday randomTimestampToday = new RandomTimestampToday();
                            tmallLogBean.setTs(randomTimestampToday.getRandomTimestampToday());

                            // Removed logId as requested
                            String currentUserId = "user_" + (100000 + random.nextInt(900000));
                            tmallLogBean.setUserId(currentUserId);
                            tmallLogBean.setOrderId("order_"+ random.nextInt(1000000));
                            tmallLogBean.setUserName(Faker.instance().name().fullName());
                            tmallLogBean.setActionType(actionTypes[random.nextInt(actionTypes.length)]);

                            Map<String, Object> actionDetail = new HashMap<>();
                            // Changed productId to shop_id and productName to shop_name
                            actionDetail.put("shop_id", "shop_" + (10000 + random.nextInt(90000)));
                            actionDetail.put("shop_name", Faker.instance().company().name());
                            actionDetail.put("category", categories[random.nextInt(categories.length)]);
                            actionDetail.put("brand", Faker.instance().company().name());
                            actionDetail.put("price", 10 + random.nextDouble() * 10000);
                            actionDetail.put("skuId", "sku_" + (10000 + random.nextInt(90000)));
                            // Added sku_name and quantity as requested
                            actionDetail.put("sku_name", Faker.instance().commerce().productName());
                            actionDetail.put("quantity", 1 + random.nextInt(100)); // Random quantity between 1-10
                            tmallLogBean.setActionDetail(actionDetail);

                            Map<String, String> deviceInfo = new HashMap<>();
                            deviceInfo.put("deviceId", "device_" + random.nextInt(100000));
                            deviceInfo.put("deviceType", deviceTypes[random.nextInt(deviceTypes.length)]);
                            deviceInfo.put("os", "Android 12");
                            deviceInfo.put("osVersion", "1.0.0");
                            deviceInfo.put("appVersion", "Tmall_10.15.0");
                            tmallLogBean.setDeviceInfo(deviceInfo);

                            Faker fakerZh = Faker.instance(new Locale("zh-CN"));
                            Map<String, String> location = new HashMap<>();
                            location.put("ip", fakerZh.internet().ipV4Address());
                            location.put("country", "中国");

                            String province = getRandomProvince();
                            String city = getRandomCity(province);
                            location.put("province", province);
                            location.put("city", city);
                            tmallLogBean.setLocation(location);

                            Map<String, String> session = new HashMap<>();
                            session.put("sessionId", "session_" + System.currentTimeMillis());
                            session.put("referer", "https://s.click.taobao.com/xxxxxx");
                            session.put("pageUrl", "https://detail.tmall.com/item.htm?id=" + (10000 + random.nextInt(90000)));
                            tmallLogBean.setSession(session);

                            ctx.collect(tmallLogBean);
                            count++;

                            Thread.sleep(10); // 可以适当减少 sleep 时间提高速度
                        }
                    }

                    @Override
                    public void cancel() {
                        isRunning = false;
                    }
                }
        );

        ObjectMapper mapper = new ObjectMapper();

        //tmallLogBeanDataStreamSource.map(mapper::writeValueAsString).print();

        KafkaUtil.producerKafka(tmallLogBeanDataStreamSource.map(mapper::writeValueAsString), "tmall_log_test");

        env.execute("TmallLogMockJob");
    }
}